# GF_PY312_CLASS_PyOllama_0_4_Calling_Ollama_0_5_Server.py
# Create by GF 2025-04-02 11:38

# Python 3 Standard Libraries.
import sqlite3
# --------------------------------------------------
# Libraries Required for Additional Functions.
# Although "tabulate 0.9.x" does not Need to "import" Separately,
# in Order to Prevent "pandas.DataFrame(...).to_markdown(...)" from Making Errors.
import tabulate
# --------------------------------------------------
import ollama
import pandas
import sqlalchemy
# ..................................................
import GF_PY3_VARIABLE_LLM_SYSTEM_PROMPT

class Additional_Functions(object):

    def Text_Opr_Content_in_Mark_Extract(self, Text:str, Mark_Start:str, Mark_End:str, With_Mark:int=1) -> str:

        # Text Operation - Extract The Content in Mark
        #
        # >>> Text = "Here is The Code:\n\n<CODE_START>\nDESCRIBE example_table;\n<CODE_END>
        # >>> print(Text)
        # Here is The Code:
        #
        # <CODE_START>
        # DESCRIBE example_table;
        # <CODE_END>
        # >>>
        # >>> Content_in_Mark = Text_Opr_Content_in_Mark_Extract(Text, "<CODE_START>\n", "<CODE_END>", 1)
        # >>> print(Content_in_Mark)
        # <CODE_START>
        # DESCRIBE example_table;
        # <CODE_END>
        # >>> Content_in_Mark = Text_Opr_Content_in_Mark_Extract(Text, "<CODE_START>\n", "<CODE_END>", 0)
        # >>> print(Content_in_Mark)
        # DESCRIBE example_table;

        Capture:list = [0, 0]
        # ..........................................
        for i in range(0, len(Text), 1):
            Slice_Idx_A = i
            Slice_Idx_B = i + len(Mark_Start)
            if (Text[Slice_Idx_A:Slice_Idx_B] == Mark_Start):
                Capture[0] = Slice_Idx_A
                break
        # ..........................................
        for i in range((Capture[0] + len(Mark_Start)), len(Text), 1):
            Slice_Idx_A = i
            Slice_Idx_B = i + len(Mark_End)
            if (Text[Slice_Idx_A:Slice_Idx_B] == Mark_End):
                Capture[1] = Slice_Idx_B
                break
        # ..........................................
        if (With_Mark == 0):
            Capture[0] = Capture[0] + len(Mark_Start)
            Capture[1] = Capture[1] - len(Mark_End)
        # ..........................................
        Content_in_Mark = Text[Capture[0]:Capture[1]]
        # ..........................................
        return Content_in_Mark

    def Text_Opr_Content_in_Mark_Delete(self, Text:str, Mark_Start:str, Mark_End:str, With_Mark:int=1) -> str:
    
        # Text Operation - Delete The Content in Mark
        #
        # >>> Text = "Here is The Code:\n\n<CODE_START>\nDESCRIBE example_table;\n<CODE_END>
        # >>> print(Text)
        # Here is The Code:
        #
        # <CODE_START>
        # DESCRIBE example_table;
        # <CODE_END>
        # >>>
        # >>> New_Text = Text_Opr_Content_in_Mark_Delete(Text, "<CODE_START>\n", "<CODE_END>", 1)
        # >>> print(new_Text)
        # Here is The Code:
        #
        # >>> New_Text = Text_Opr_Content_in_Mark_Delete(Text, "<CODE_START>\n", "<CODE_END>", 0)
        # >>> print(New_Text)
        # Here is The Code:
        #
        # <CODE_START>
        # <CODE_END>
    
        Capture:list = [0, 0]
        # ..........................................
        for i in range(0, len(Text), 1):
            Slice_Idx_A = i
            Slice_Idx_B = i + len(Mark_Start)
            if (Text[Slice_Idx_A:Slice_Idx_B] == Mark_Start):
                Capture[0] = Slice_Idx_A
                break
        # ..........................................
        for i in range((Capture[0] + len(Mark_Start)), len(Text), 1):
            Slice_Idx_A = i
            Slice_Idx_B = i + len(Mark_End)
            if (Text[Slice_Idx_A:Slice_Idx_B] == Mark_End):
                Capture[1] = Slice_Idx_B
                break
        # ..........................................
        if (With_Mark == 0):
            Capture[0] = Capture[0] + len(Mark_Start)
            Capture[1] = Capture[1] - len(Mark_End)
        # ..........................................
        New_Text = Text.replace(Text[Capture[0]:Capture[1]], '')
        # ..........................................
        return New_Text

    def Text_Opr_Loop_Read_by_Word_Num(self, Text:str, Begin_Index:int, Word_Num:int) -> tuple:

        # Text Operation - Loop Read by Word Number
        #
        # >>> Text = "A string is a string of characters consisting of numbers, letters, and underscores."
        # >>> Text_Readed_1 = Text_Opr_Loop_Read_by_Word_Num(Text,   0, 10)
        # >>> print(Text_Readed_1)
        # (10, "A string i")
        # >>> Text_Readed_2 = Text_Opr_Loop_Read_by_Word_Num(Text,  10, 20)
        # >>> print(Text_Readed_2)
        # (30, "s a string of charac")
        # >>> Text_Readed_3 = Text_Opr_Loop_Read_by_Word_Num(Text, 107, 10)
        # >>> print(Text_Readed_3)
        # (None, '')

        Text_Total_Word_Num = len(Text)
        # ..............................................
        Bgn_Idx = Begin_Index
        End_Idx = Bgn_Idx + Word_Num # Python 3 字符串切片, 含开始索引, 不含结束索引
        # ..............................................
        if (End_Idx >= Text_Total_Word_Num):
            End_Idx = Text_Total_Word_Num
        # ..............................................
        Readed_Text_Slice = Text[Bgn_Idx:End_Idx]
        # ..............................................
        Bgn_Idx = End_Idx           # 设置 Next Loop 开始索引
        # ..............................................
        if (End_Idx >= Text_Total_Word_Num):
            Bgn_Idx = None
        # ..............................................
        return (Bgn_Idx, Readed_Text_Slice)

    def Pandas_2_x_DataFrame_Preview(self, Pandas_DataFrame):

        # Requirement: pandas 2.x.x
        #
        # Example:
        #
        # >>> print(df)
        #    A  B  C  D  E  F  G  H  I
        # 0  1  1  1  1  1  1  1  1  1
        # 1  2  2  2  2  2  2  2  2  2
        # 2  3  3  3  3  3  3  3  3  3
        # 3  4  4  4  4  4  4  4  4  4
        # 4  5  5  5  5  5  5  5  5  5
        # 5  6  6  6  6  6  6  6  6  6
        # 6  7  7  7  7  7  7  7  7  7
        # 7  8  8  8  8  8  8  8  8  8
        # 8  9  9  9  9  9  9  9  9  9
        #
        # >>> df = Pandas_2_x_DataFrame_Preview(df)
        # >>> print(df)
        #        A    B    C    D  ...    F    G    H    I
        #   0    1    1    1    1  ...    1    1    1    1
        #   1    2    2    2    2  ...    2    2    2    2
        #   2    3    3    3    3  ...    3    3    3    3
        #   3    4    4    4    4  ...    4    4    4    4
        # ...  ...  ...  ...  ...  ...  ...  ...  ...  ...
        #   5    6    6    6    6  ...    6    6    6    6
        #   6    7    7    7    7  ...    7    7    7    7
        #   7    8    8    8    8  ...    8    8    8    8
        #   8    9    9    9    9  ...    9    9    9    9

        Rows_Num = Pandas_DataFrame.shape[0]
        Cols_Num = Pandas_DataFrame.shape[1]

        Result_DF = Pandas_DataFrame.copy()
        if (Rows_Num > 8 and "..." not in Result_DF.index):
            Res_Rows_Idx_A   = Pandas_DataFrame.index[0:4].tolist()
            Res_Rows_Idx_B   = Pandas_DataFrame.index[-4:].tolist()
            Res_Rows_Idx_All = Res_Rows_Idx_A + ["..."] + Res_Rows_Idx_B
            # ......................................
            Result_DF_A = Result_DF.iloc[Res_Rows_Idx_A, :]
            Result_DF_B = Result_DF.iloc[Res_Rows_Idx_B, :]
            # ......................................
            Empty_Row = {Key:"..." for Key in Result_DF.columns}
            Result_DF = pandas.concat([Result_DF_A, pandas.DataFrame([Empty_Row]), Result_DF_B])
            # ......................................
            Result_DF.index = Res_Rows_Idx_All

        if (Cols_Num > 8 and "..." not in Result_DF.columns):
            Res_Cols_Name_A   = Pandas_DataFrame.columns[0:4].tolist()
            Res_Cols_Name_B   = Pandas_DataFrame.columns[-4:].tolist()
            Res_Cols_Name_All = Res_Cols_Name_A + ["..."] + Res_Cols_Name_B
            # ......................................
            Result_DF["..."] = "..."
            # ......................................
            Result_DF = Result_DF[Res_Cols_Name_All]

        return Result_DF

    def MySQL_Query(self, SQL_Statment:str):

        SQLAlchemy_Engine = sqlalchemy.create_engine("mysql+pymysql://robot:12345678@127.0.0.1/working")
        # ..........................................
        Result_DF = pandas.read_sql_query(SQL_Statment, con=SQLAlchemy_Engine)
        # ..........................................
        return Result_DF

    def SQLite3_Query(self, SQL_Statment:str):

        Conn = sqlite3.connect('llm_prompt.db') # 连接到 SQLite 3 数据库
        # ..........................................
        Cursor = Conn.cursor()
        # ..........................................
        Cursor.execute(SQL_Statment)            # 查询数据
        # ..........................................
        Rows = Cursor.fetchall()
        # ..........................................
        Conn.close()
        # ..........................................
        return Rows

class PyOllama_0_4_Calling_Ollama_0_5_Server(Additional_Functions):

    def __init__(self):

        self.OLLAMA_MODEL = "qwen2.5-coder:3b"
        self.PYTHON_3_EXEC_MYSQL_STATMENT        = GF_PY3_VARIABLE_LLM_SYSTEM_PROMPT.PYTHON_3_EXEC_MYSQL_STATMENT
        self.SYSTEM_PROMPT_DATABASE_TASK_TYPE    = GF_PY3_VARIABLE_LLM_SYSTEM_PROMPT.LLM_SYSTEM_PROMPT_DATABASE_TASK_TYPE
        self.SYSTEM_PROMPT_DATABASE_OVERVIEW     = GF_PY3_VARIABLE_LLM_SYSTEM_PROMPT.LLM_SYSTEM_PROMPT_DATABASE_OVERVIEW
        self.SYSTEM_PROMPT_DATABASE_TABLE_CHOICE = GF_PY3_VARIABLE_LLM_SYSTEM_PROMPT.LLM_SYSTEM_PROMPT_DATABASE_TABLE_CHOICE
        self.SYSTEM_PROMPT_DATABASE_TABLE_QUERY  = GF_PY3_VARIABLE_LLM_SYSTEM_PROMPT.LLM_SYSTEM_PROMPT_DATABASE_TABLE_QUERY

    def Chat_for_Single_Round(self, Message:str):

        Messages = [{"role": "user", "content": Message}]
        # ..........................................
        # 调用 Ollama 模型生成回复
        Model_Response = ollama.chat(model=self.OLLAMA_MODEL, messages=Messages)
        # ..........................................
        # 获取模型生成内容: Model_Response.get("message").get("content")
        return Model_Response

    def Chat_for_Multiple_Rounds(self, Message_List:str):

        # Message_List 应包含:
        # [
        #     {"role":      "user", "content": "Hello!"},
        #     {"role": "assistant", "content": "Hello, I am an AI assistant!"},
        #     {"role":      "user", "content": "What time is it?"},
        #     ......
        # ]
        # ..........................................
        # 调用 Ollama 模型生成回复
        Model_Response = ollama.chat(model=self.OLLAMA_MODEL, messages=Message_List)
        # ..........................................
        # 获取模型生成内容: Model_Response.get("message").get("content")
        return Model_Response

    def AI_Agent_for_Database(self, Message:str):

        # Requirement: tabulate 0.9.x
        # Although "tabulate 0.9.x" does not Need to "import" Separately,
        # in Order to Prevent "pandas.DataFrame(...).to_markdown(...)" from Making Errors.

        # ##########################################
        # Ollama 调用模型进行第 1 轮对话
        # ##########################################

        # 第 1 轮对话 (Prompt 预处理)
        PLACE_HOLDER_GOAL = Message
        COMPLETE_PROMPT = self.SYSTEM_PROMPT_DATABASE_TASK_TYPE # 完整的提示词 (替换其中的 Place Holder 占位符)
        COMPLETE_PROMPT = COMPLETE_PROMPT.replace("::GOAL::", PLACE_HOLDER_GOAL)
        # ..........................................
        Message_List = [{"role": "user", "content": COMPLETE_PROMPT}] # 第 1 轮对话 (完整的 Prompt)
        Model_Rsp = self.Chat_for_Multiple_Rounds(Message_List)       # 第 1 轮对话 (让 Model 响应)
        Model_Ctt = Model_Rsp.get("message").get("content")
        # ..........................................
        TASK_TYPE = self.Markdown_Extract_Code_Block_from_String(Model_Ctt)

        try:
            if   (TASK_TYPE == "1"):
                print("[Message] Executing Overview Task for Database Table...\n")

                # ##################################
                # Ollama 调用模型进行第 2 轮对话
                # ##################################

                # 第 1 轮对话 (Prompt 预处理)
                PLACE_HOLDER_GOAL = Message
                COMPLETE_PROMPT = self.SYSTEM_PROMPT_DATABASE_OVERVIEW # 完整的提示词 (替换其中的 Place Holder 占位符)
                COMPLETE_PROMPT = COMPLETE_PROMPT.replace("::GOAL::", PLACE_HOLDER_GOAL)
                # ..................................
                Message_List.append({"role": "assistant", "content":Model_Ctt})   # 第 2 轮对话 (添加 Historys)
                Message_List.append({"role": "user", "content": COMPLETE_PROMPT}) # 第 2 轮对话 (完整的 Prompt)
                Model_Rsp = self.Chat_for_Multiple_Rounds(Message_List)           # 第 2 轮对话 (让 Model 响应)
                Model_Ctt = Model_Rsp.get("message").get("content")
                # ..................................
                SQL_STATMENT = self.Markdown_Extract_Code_Block_from_String(Model_Ctt)

                # ##################################
                # 执行 Ollama 中模型生成的 SQL 语句
                # ##################################

                Result = {"role": "assistant", "content": "AI 数据管理员开小差了, 请稍后重试..."}

                if (SQL_STATMENT == None):
                    print("[Warning] SQL Block Generation Failed in Overview Task.")
                    # ..............................
                    Result_JSON = """{"role": "assistant", "content": "对不起, 我刚刚忘记数据库语句的语法了, 麻烦亲再问一遍"}"""
                    # ..............................
                    return Result_JSON
                else:
                    print("[Message] SQL Block Will Executed in Overview Task:\n", SQL_STATMENT)
                    # ..............................
                    CODE_BLOCK_TEXT =self.PYTHON_3_EXEC_MYSQL_STATMENT.replace("::SQL_STATMENT::", SQL_STATMENT)
                    # ..............................
                    Executed = {} # 定义一个字典来存储 exec 中的变量
                    exec(CODE_BLOCK_TEXT, Executed)
                    # ..............................
                    Pandas_DF_to_JSON = Executed["Result_DF"].to_json(force_ascii=False, orient="records")
                    Result_JSON = """{"role": "assistant", "content": %s}""" % Pandas_DF_to_JSON
                    # ..............................
                    return Result_JSON

            elif (TASK_TYPE == "2"):
                print("[Message] Executing Query Task for Database Table...\n")

                # ##################################
                # Ollama 调用模型进行第 2 轮对话
                # ##################################

                # 第 2 轮对话 (Prompt 预处理)
                PLACE_HOLDER_GOAL = Message
                COMPLETE_PROMPT = self.SYSTEM_PROMPT_DATABASE_TABLE_CHOICE # 完整的提示词 (替换其中的 Place Holder 占位符)
                COMPLETE_PROMPT = COMPLETE_PROMPT.replace("::GOAL::", PLACE_HOLDER_GOAL)
                # ..................................
                Message_List.append({"role": "assistant", "content":Model_Ctt})   # 第 2 轮对话 (添加 Historys)
                Message_List.append({"role": "user", "content": COMPLETE_PROMPT}) # 第 2 轮对话 (完整的 Prompt)
                Model_Rsp = self.Chat_for_Multiple_Rounds(Message_List)           # 第 2 轮对话 (让 Model 响应)
                Model_Ctt = Model_Rsp.get("message").get("content")
                # ..................................
                TABLE_NAME = self.Markdown_Extract_Code_Block_from_String(Model_Ctt)

                # ##################################
                # Ollama 调用模型进行第 3 轮对话
                # ##################################

                # 第 3 轮对话 (Prompt 预处理)
                Table_Desc_DF = self.MySQL_Query("DESCRIBE %s;"      % TABLE_NAME) # 数据表描述 (DESCRIBE)
                Table_Data_DF = self.MySQL_Query("SELECT * FROM %s;" % TABLE_NAME) # 数据表数据 (SELECT *)
                # ..................................
                Table_Data_DF = self.Pandas_2_x_DataFrame_Preview(Table_Data_DF) # 只输出 8 x 8 的 DataFrame 预览
                # ..................................
                PLACE_HOLDER_TABLE_NAME     = TABLE_NAME
                PLACE_HOLDER_TABLE_DESCRIBE = Table_Desc_DF.to_markdown(index=False)
                PLACE_HOLDER_TABLE_SAMPLE   = Table_Data_DF.to_markdown(index=False)
                PLACE_HOLDER_GOAL           = Message
                # ..................................
                COMPLETE_PROMPT = self.SYSTEM_PROMPT_DATABASE_TABLE_QUERY # 完整的提示词 (替换其中的 Place Holder 占位符)
                COMPLETE_PROMPT = COMPLETE_PROMPT.replace("::TABLE_NAME::", PLACE_HOLDER_TABLE_NAME)
                COMPLETE_PROMPT = COMPLETE_PROMPT.replace("::TABLE_DESCRIBE::", PLACE_HOLDER_TABLE_DESCRIBE)
                COMPLETE_PROMPT = COMPLETE_PROMPT.replace("::TABLE_SAMPLE::", PLACE_HOLDER_TABLE_SAMPLE)
                COMPLETE_PROMPT = COMPLETE_PROMPT.replace("::GOAL::", PLACE_HOLDER_GOAL)
                # ..................................
                Message_List.append({"role": "assistant", "content":Model_Ctt})   # 第 3 轮对话 (添加 Historys)
                Message_List.append({"role": "user", "content": COMPLETE_PROMPT}) # 第 3 轮对话 (完整的 Prompt)
                Model_Rsp = self.Chat_for_Multiple_Rounds(Message_List)           # 第 3 轮对话 (让 Model 响应)
                Model_Ctt = Model_Rsp.get("message").get("content")
                # ..................................
                SQL_STATMENT = self.Markdown_Extract_Code_Block_from_String(Model_Ctt)

                # ##################################
                # 执行 Ollama 中模型生成的 SQL 语句
                # ##################################

                if (SQL_STATMENT == None):
                    print("[Warning] SQL Block Generation Failed in Query Task.")
                    # ..............................
                    Result_JSON = """{"role": "assistant", "content": "对不起, 我刚刚忘记数据库语句的语法了, 麻烦亲再问一遍"}"""
                    # ..............................
                    return Result_JSON
                else:
                    print("[Message] SQL Block Will Executed in Query Task:\n", SQL_STATMENT)
                    # ..............................
                    CODE_BLOCK_TEXT =self.PYTHON_3_EXEC_MYSQL_STATMENT.replace("::SQL_STATMENT::", SQL_STATMENT)
                    # ..............................
                    Executed = {} # 定义一个字典来存储 exec 中的变量
                    exec(CODE_BLOCK_TEXT, Executed)
                    # ..............................
                    Pandas_DF_to_JSON = Executed["Result_DF"].to_json(force_ascii=False, orient="records")
                    Result_JSON = """{"role": "assistant", "content": %s}""" % Pandas_DF_to_JSON
                    # ..............................
                    return Result_JSON
            else:
                Result_JSON = """{"role": "assistant", "content": "对不起, 我刚刚没有理解到你的意图, 麻烦亲在问题中多提供一些细节"}"""
                # ..................................
                return Result_JSON

        except Exception as e:
            print(e)
            # ......................................
            Result_JSON = """{"role": "assistant", "content": "服务器繁忙, 请稍后再试"}"""
            # ......................................
            return Result_JSON

    def AI_Agent_for_Office_PPT(self, Message:str, Save_Path:str):
    
        COMPLETE_PROMPT = self.SQLite3_Query("SELECT prompt FROM llm_system_prompt_office_ppt WHERE id = 1;")[0][0]
        COMPLETE_PROMPT = COMPLETE_PROMPT.replace("::SAVE_PATH::", Save_Path)
        COMPLETE_PROMPT = COMPLETE_PROMPT.replace("::GOAL::", Message)
        # debug_01 = COMPLETE_PROMPT
        # ..........................................
        Chat_Response = self.Chat_for_Single_Round(COMPLETE_PROMPT)
        Chat_Response_Content = Chat_Response.get("message").get("content")
        # debug_02 = Chat_Response_Content
        # ..........................................
        TEXT_CODE_BLOCK = self.Text_Opr_Content_in_Mark_Extract(Chat_Response_Content, "```python\n", "```", 0)
        TEXT_IDEA_BLOCK = self.Text_Opr_Content_in_Mark_Delete(Chat_Response_Content, "```python\n", "```", 1)
        # ..........................................
        TEXT_IDEA_BLOCK = TEXT_IDEA_BLOCK.replace('\"', '')    # 移除 (") 双引号
        TEXT_IDEA_BLOCK = TEXT_IDEA_BLOCK.replace('\'', '')    # 移除 (') 单引号
        TEXT_IDEA_BLOCK = TEXT_IDEA_BLOCK.replace('\n', "\\n") # 替换 (\n) 为 (\\n)
        # ..........................................
        try:
            exec(TEXT_CODE_BLOCK)
            # ......................................
            debug_03 = "null"
        except Exception as e:
            print(e)
            # ......................................
            debug_03 = e
        # ..........................................
        return """{"role": "assistant", "content": %s, "debug": %s}""" % (TEXT_IDEA_BLOCK, debug_03)

# EOF Signed by GF.
